package com.niit.rdd

import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

object Spark_RDD_Transform02 {

  def main(args: Array[String]): Unit = {

    //准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")
    /*
    filter:将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。 当
      数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出
      现数据倾斜。
      需求：获得列表中的奇数
     */
    val rdd1 = sc.makeRDD( List(1,2,3,4) )
    val fliterRDD =  rdd1.filter(num => num%2 !=0 )
    fliterRDD.collect().foreach(println)


    /*
    distinct:将数据集中重复的数据去重
     */
    val rdd2 = sc.makeRDD(List(1,2,3,4,1,2,3,4,5))
    val disRdd =  rdd2.distinct(2)
    println("分区数：",disRdd.getNumPartitions)
    disRdd.collect().foreach(println)
    println("------------------------------")
    /*
    sortBy:默认是升序 如果想改成降序，第二个参数设置为false
     */
    val sortRdd =  disRdd.sortBy(num=>num,false)
    sortRdd.collect().foreach(println)

    //按照key 进行排序
    val rdd3 =  sc.makeRDD( List( ("1","a") ,("3","c") , ("2","b") ) )
    val sortRdd2 = rdd3.sortBy(t=> t._1.toInt)
    sortRdd2.collect().foreach(println)

    /*
     coalesce
    根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率
    当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减
    少 分区的个数，减小任务调度成本
     */
    val rdd4 = sc.makeRDD( List(1,2,3,4,5,6),3 )//【1,2】 【3,4】 【5,6】==> 【1,2,3】 【4,5,6】
    //减少分区会发生数据倾斜 ，如果去避免数据倾斜呢？ 将第二个参数设置true,只能避免数据区倾斜的问题
    val coaRdd =  rdd4.coalesce(2,true)
    //coaRdd.saveAsTextFile("ouput")

    /*
      repatition
      该操作内部其实执行的是 coalesce 操作，参数 shuffle 的默认值为 true。无论是将分区数
    多的 RDD 转换为分区数少的 RDD，还是将分区数少的 RDD 转换为分区数多的 RDD，
      repartition 操作都可以完成，因为无论如何都会经 shuffle 过程。
     */
    val rdd5 = sc.makeRDD( List(1,2,3,4,5,6),2 )
    val reRdd =  rdd5.repartition(3)
    //reRdd.saveAsTextFile("ouput1")

    /*
      交集 并集 差集 拉链
     */
    //交集
    val rdd6 =  sc.makeRDD( List(1,2,3,4))
    val rdd7 =  sc.makeRDD( List(3,4,5,6) )
    val inRdd =  rdd6.intersection(rdd7)
    println(inRdd.collect().mkString(","))

    //并集
    val unRdd =  rdd6.union(rdd7)
    println(unRdd.collect().mkString(","))

    //差集
    val subRdd =  rdd6.subtract(rdd7)
    println(subRdd.collect().mkString(","))

    //拉链
    val zipRdd =  rdd6.zip(rdd7)
    println(zipRdd.collect().mkString(","))

    /*
    . partitionBy 将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitione
     */
    val rdd8 = sc.makeRDD( List(1,2,3,4),2 )
    //隐式转换
    val mapRdd = rdd8.map( (_,1) )//=>(1,1) (2,1) (3,1) (4,1)
    val parRdd = mapRdd.partitionBy( new HashPartitioner( 3 ) )
    //parRdd.saveAsTextFile("output2")

    /*
    reduceByKey:可以将数据按照相同的 Key 对 Value 进行聚合
     */
    val rdd9 = sc.makeRDD( List( ("a",1) ,("a",2) ,("b",3) ,("b",4)) )
      /*
        a   x=1   y =2
        b   x=3   y=4
       */
    //reduceByKey 是根据key进行聚合 所以key是不参与计算
    val redRdd =  rdd9.reduceByKey( (x:Int,y:Int)=>{
          println(s"x=${x},y=${y}")
          x + y
      } )
    redRdd.collect().foreach(println)


    //关闭环境
    sc.stop()

  }

}
